package graphx

import algorithm.ShortestPathAlgorithm
import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph, lib}
import org.apache.spark.{SparkConf, SparkContext}

object bfsExample extends App{
  //设置日志级别
  Logger.getLogger("org").setLevel(Level.ERROR)

  val conf = new SparkConf().setAppName("graph-bfs-scala").setMaster("local[2]")
  val sc = new SparkContext(conf)

  //Dijkstra 算法是在图中计算从一个指定的顶点到其他每一个顶点的路径距离
  val myVertices = sc.makeRDD(Array((1L,"A"),(2L ,"B"),( 3L,"C"),
    (4L,"D"),(5L,"E"),(6L,"F"),(7L,"G")))


  val myEdges = sc.makeRDD(
    Array(Edge(1L, 2L, 7.0),
      Edge(1L, 4L, 5.0),
      Edge(2L, 3L, 8.0),
      Edge(2L, 4L, 9.0),
      Edge(2L, 5L , 7.0),
      Edge(3L, 5L, 5.0),
      Edge(4L, 5L , 15.0),
      Edge(4L, 6L , 6.0 ),
      Edge(5L, 6L, 8.0),
      Edge(5L, 7L, 9.0),
      Edge( 6L, 7L , 11.0)))

  val myGraph = Graph(myVertices, myEdges)

  //Note graphx自带的算法 其返回的是各个节点到根节点的路径...
  //dijkstra 其返回的是根节点到各个节点的路径

  //利用graphx自带的算法
  lib.ShortestPaths.run(myGraph,Array(3L)).vertices.foreach(x=>println(x))

  //利用dijkstra最短路径计算
  val bfsGraph = myGraph.mapTriplets(_=>1D) //将所有边的权重设为1
    ShortestPathAlgorithm.dijkstraWithTrace(bfsGraph,3L,direct = true).vertices.collect().foreach(data
    =>{
      println(data._1,data._2)
    })
}
